#include <vector>
#include <boost/bind.hpp>
#include <sstream>
#include <iostream>
#include <boost/type_traits.hpp>
#include <boost/format.hpp>

#include "con_connection.h"
#include "log.h"
#include "options.h"
#include "con_command.h"
#include "options.h"
#include "controllers_network.h"

con_connection::con_connection(boost::asio::io_service &_io_service, controllers_network &_network, sqlhandle &_sqlhandle):
    pult_connection_(false),
    controller_id_(0),
    secure_connection_(false),
    need_remove_controller_(false),
    timer_(_io_service),
    socket_(_io_service),
	strand_(_io_service),
	command_(NULL),
	network_(_network),
	sqlhandle_(_sqlhandle)
{
}

con_connection::~con_connection()
{
}

void con_connection::close_connection()
{
  timer_.cancel();
  
  try
  {
	socket_.cancel();
	socket_.close();
  }	
  catch(...)
  {
  }
  
  if (need_remove_controller_)
	{
	  network_.remove_controller_connection(controller_id_);
	  sqlhandle_.delete_controller_from_network(controller_id_);
	}
	
  std::string con = pult_connection_ ?  "pult" : boost::lexical_cast<std::string>(controller_id_);

  if (controller_id_ != 0)
	  g_log.msg(MSG_NORMAL,boost::format("[%s]: Disconnected.") % con);

  if (command_)
  {
	delete command_;
	command_ = NULL;
  }
}

boost::asio::ip::tcp::socket& con_connection::socket()
{
    return socket_;
}

void con_connection::start()
{
	start_new_command(new hello_command(network_, sqlhandle_));

	start_command();
}

void con_connection::controller_handle_read(const boost::system::error_code& _error, size_t _bytes_transferred)
{
   timer_.cancel();
   
   if (!_error && (_bytes_transferred > 0))
   {
  	   read_restart_count_ = 0;
  	   
	   write_buffer_.clear();

	   dump_packet(true, boost::asio::buffer(read_buffer_), _bytes_transferred);
	   
	   	if (secure_connection_)
		{
		  if (!encoder_.decode((unsigned char*)read_buffer_.data(), _bytes_transferred))
		  {
			g_log.msg(MSG_NORMAL, boost::format("[%d] connect: status='Invalid packet size: %d'") % controller_id_ % _bytes_transferred);
			start_queue_command();
			return;
		  }
		}
		
		dump_packet(true, boost::asio::buffer(read_buffer_), _bytes_transferred);
		
	   con_command::return_t val = command_->process_request(boost::asio::const_buffer(boost::asio::buffer(read_buffer_)), _bytes_transferred, write_buffer_, shared_from_this());

	   switch(val)
	   {
		   case con_command::start_write:

			   dump_packet(false, boost::asio::buffer(write_buffer_[0]));

			   boost::asio::async_write(socket_,
						write_buffer_,
						strand_.wrap(boost::bind(&con_connection::controller_handle_write, shared_from_this(),
						  boost::asio::placeholders::error)));
			   break;

		   case con_command::start_last_write:
			   boost::asio::async_write(socket_,
						write_buffer_,
						strand_.wrap(boost::bind(&con_connection::controller_handle_last_write, shared_from_this(),
						  boost::asio::placeholders::error)));
			   break;

		   case con_command::start_timer:
				   start_queue_command();
			   break;
	   }
   }
   else
   {
	  if (_error !=  boost::asio::error::operation_aborted)
  		close_connection();
   }
}

void con_connection::start_command()
{
	write_buffer_.clear();

	if (command_->create_command(write_buffer_, shared_from_this()))
	{
	    dump_packet(false, boost::asio::buffer(write_buffer_[0]));

		boost::asio::async_write(socket_,
					write_buffer_,
					strand_.wrap(boost::bind(&con_connection::controller_handle_write, shared_from_this(),
					  boost::asio::placeholders::error)));
	}
	else
	{

		if (!pult_connection_)
	  	  start_read_timeout();

		socket_.async_read_some(boost::asio::buffer(read_buffer_),
		        strand_.wrap(boost::bind(&con_connection::controller_handle_read, shared_from_this(),
		          boost::asio::placeholders::error,
		          boost::asio::placeholders::bytes_transferred)));
	}
}

void con_connection::controller_handle_write(const boost::system::error_code& _error)
{
	if (!_error)
	{
		if (command_->need_read())
		{
			start_new_command(command_->get_next_command());

			if (!pult_connection_)
			  start_read_timeout();

			socket_.async_read_some(boost::asio::buffer(read_buffer_),
					strand_.wrap(boost::bind(&con_connection::controller_handle_read, shared_from_this(),
					  boost::asio::placeholders::error,
					  boost::asio::placeholders::bytes_transferred)));
		}
		else
		{
			start_queue_command();
		}
	}
	else
	{
	  if (_error !=  boost::asio::error::operation_aborted)
		close_connection();
	}
}

void con_connection::start_read_timeout()
{
		timer_.expires_from_now(boost::posix_time::seconds(10));

		timer_.async_wait(strand_.wrap(boost::bind(&con_connection::controller_handle_read_timeout,
				shared_from_this(), boost::asio::placeholders::error)));

}

void con_connection::controller_handle_last_write(const boost::system::error_code& _error)
{
    boost::system::error_code ignored_ec;
    socket_.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ignored_ec);
    
    if (_error !=  boost::asio::error::operation_aborted)
  	  close_connection();
}

void con_connection::controller_handle_timer( const boost::system::error_code &_error)
{
	if(start_command_from_queue())
	{
		start_command();
	}
	else
	{
		if (!_error)
		{
			start_new_command(command_->get_next_command());

			start_command();
		}
	}
}

bool con_connection::start_command_from_queue()
{
	boost::mutex::scoped_lock(command_queue_mutex_);

	if (command_queue_.size() > 0)
	{
		con_command *cmd = command_queue_.front();

		command_queue_.pop_front();

		start_new_command(cmd);

		return true;
	}
	else
	{
		return false;
	}
}

void con_connection::start_queue_command()
{
	if (start_command_from_queue())
	{
		socket_.get_io_service().post(
				strand_.wrap(boost::bind(&con_connection::start_command, shared_from_this()))
				);
	}
	else
	{
		timer_.expires_from_now(boost::posix_time::seconds(g_config.m_rescan_time));

		timer_.async_wait(strand_.wrap(boost::bind(&con_connection::controller_handle_timer,
				shared_from_this(), boost::asio::placeholders::error)));
	}
}

void con_connection::start_new_command(con_command *_command)
{
		if (_command == NULL)
		{
		  close_connection();
		  
		  return;
		}  
		else if (command_ != _command)
		{
			delete command_;
		}

		command_ = _command;
}

void con_connection::add_command_to_queue(con_command *_command)
{
	boost::mutex::scoped_lock(command_queue_mutex_);

	command_queue_.push_back(_command);

	timer_.cancel();
}

void con_connection::dump_packet(bool _read, boost::asio::const_buffer _buffer, unsigned int _size)
{
	const unsigned char *src = boost::asio::buffer_cast<const unsigned char*>(_buffer);
	unsigned int src_len = (_size > 0) ? _size : boost::asio::buffer_size(_buffer);

    if (!g_config.m_dump_all_packet) 
    	return;

    std::string read = _read ? "read:" : "write";
    std::string con = pult_connection_ ?  str(boost::format("[pult] %s") % read) :
					  str(boost::format("[%d] %s") % controller_id_ % read);

    g_log.msg(MSG_NORMAL, boost::format("%s '%s'") % con % con_command::dump_packet(src, src_len));
}

void con_connection::controller_handle_read_timeout(const boost::system::error_code &_error)
{
  if (!_error)
  {
	std::string con = pult_connection_ ?  "[pult]" :
					  (str(boost::format("[%d]") % controller_id_));
					  
	g_log.msg(MSG_NORMAL, boost::format("%s: %s") % con % "Read timeout !");
	
	if (read_restart_count_ >= 3)
	{
		close_connection();
	}
	else
	{
	    socket_.cancel();
		read_restart_count_++;
		start_command();
	}
  }
}
